package com.heroh.kettle;

import com.heroh.kettle.entity.ParamDbConnection;
import com.heroh.kettle.util.TransBuild;
import com.heroh.kettle.util.MetaBuild;
import com.heroh.kettle.vo.InsertUpdateVO;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.insertupdate.InsertUpdateMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;

import java.util.Arrays;

@Slf4j
public class DBTransferUtil {

    private ParamDbConnection srcDb;

    private ParamDbConnection destDb;

    private TransMeta meta;

    public DBTransferUtil(ParamDbConnection srcDb, ParamDbConnection destDb) {
        this.srcDb = srcDb;
        this.destDb = destDb;
        this.meta = new MetaBuild()
                .setConns(Arrays.asList(srcDb,destDb))
                .setTransLog()
                .build();
    }

    /**
     * 表输入输出
     * @param fromSql
     * @param destTable
     */
    public void transTableBySql(String fromSql,String destTable) {
        StepMeta step1 = setTableInputStep(srcDb.getDbConnectionCode() , fromSql, "table_input");
        StepMeta step2 = setTableOutput(destDb.getDbConnectionCode(), destTable, "table_insert");
        meta.addTransHop(new TransHopMeta(step1, step2));
        meta.setName("trans-"+destTable);
        runTrans(meta);
    }

    /**
     *  设置更新或插入步骤
     *
     * @return StepMeta 表输出步骤
     * @author jianshengxiong
     * @date 2019/12/31/15:45
     */
    public StepMeta setInsertUpdateMeta(InsertUpdateVO insertUpdateVO) {
        // 创建插入与更新
        InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
        String pluginId = PluginRegistry.getInstance().getPluginId(StepPluginType.class, insertUpdateMeta);
        // 配置目标数据库配置名
        DatabaseMeta database_target = meta.findDatabase(insertUpdateVO.getDestDb());
        insertUpdateMeta.setDatabaseMeta(database_target);
        // 设置目标表名
        insertUpdateMeta.setTableName(insertUpdateVO.getDestTable());
        // 设置用来查询的关键字
        insertUpdateMeta.setKeyLookup(insertUpdateVO.getKeyLoop());
        insertUpdateMeta.setKeyStream(insertUpdateVO.getKeyStream());
        insertUpdateMeta.setKeyStream2(insertUpdateVO.getKeyStream2());// 这一步不能省略
        insertUpdateMeta.setKeyCondition(insertUpdateVO.getKeyCondition());
        // 设置要更新的字段
        insertUpdateMeta.setUpdateLookup(insertUpdateVO.getUpdateLookup());
        insertUpdateMeta.setUpdateStream(insertUpdateVO.getUpdateStream());
        insertUpdateMeta.setUpdate(insertUpdateVO.getUpdateOrNot());
        // 添加步骤到转换中
        StepMeta stepMeta = new StepMeta(pluginId, insertUpdateVO.getDestTable(), insertUpdateMeta);
        meta.addStep(stepMeta);
        return stepMeta;
    }

    /**
     *  设置表输入
     *
     * @return StepMeta 表输入步骤
     * @author jianshengxiong
     * @date 2019/12/31/15:45
     */
    private StepMeta setTableInputStep(String dbName, String sql, String stepName) {
        // 创建表输入
        TableInputMeta tableInputMeta = new TableInputMeta();
        // 指定数据源数据库配置名
        DatabaseMeta source = meta.findDatabase(dbName);
        tableInputMeta.setDatabaseMeta(source);
        tableInputMeta.setSQL(sql);

        String pluginId = PluginRegistry.getInstance().getPluginId(StepPluginType.class, tableInputMeta);
        // 将表输入添加到转换中
        StepMeta stepMeta = new StepMeta(pluginId, stepName, tableInputMeta);
        meta.addStep(stepMeta);
        return stepMeta;
    }

    /**
     *  设置表输出步骤
     *
     * @return StepMeta 表输出步骤
     * @author jianshengxiong
     * @date 2019/12/31/15:45
     */
    private StepMeta setTableOutput(String dbName, String tableName, String stepName) {
        // 创建表输出
        TableOutputMeta tableOutputMeta = new TableOutputMeta();
        // 配置表输出的目标数据库配置名
        DatabaseMeta targetDb = meta.findDatabase(dbName);
        tableOutputMeta.setDatabaseMeta(targetDb);
        tableOutputMeta.setTableName(tableName);

        String pluginId = PluginRegistry.getInstance().getPluginId(StepPluginType.class, tableOutputMeta);
        // 将表输出添加到转换中
        StepMeta stepMeta = new StepMeta(pluginId, stepName, tableOutputMeta);
        meta.addStep(stepMeta);
        return stepMeta;
    }

    private void runTrans(TransMeta transMeta) {
        try {
            Trans trans = new TransBuild().build(transMeta);
            trans.execute(null);
            trans.waitUntilFinished();
            if (trans.getErrors() > 0) {
                log.error("kettle转换异常");
            } else {
                log.info("执行成功");
            }
        } catch (KettleException e) {
            log.error("kettle配置异常");
        }
    }
}
